package com.atguigu.flink.sql.window;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * Created by Smexy on 2023/4/14
 *
 *  把 state/Demo11_TopN 使用sql实现
 */
public class Demo8_TopN
{
    public static void main(String[] args) {

        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tableEnvironment = TableEnvironment.create(environmentSettings);

        //1.读取数据，封装为表
        String createTableSql = " create table t1 ( uid BIGINT, itemId BIGINT , cid INT , behavior STRING, ts BIGINT ," +
            "   pt  AS PROCTIME() , " +
            "   et  AS TO_TIMESTAMP_LTZ(ts,0)  ," +
            "   WATERMARK FOR et AS et - INTERVAL '0.001' SECOND " +
            "    )WITH (" +
            "  'connector' = 'filesystem'," +
            "  'path' = 'data/UserBehavior.csv'," +
            "  'format' = 'csv'" +
            ") ";

        //建表
        tableEnvironment.executeSql(createTableSql);

        //2.滑动窗口聚合
        String hopWindowSql = "SELECT itemId, window_start, window_end, count(*) click " +
            "  FROM TABLE(" +
            "    HOP(TABLE t1, DESCRIPTOR(et),  INTERVAL '5' MINUTES, INTERVAL '1' HOUR)" +
            "  )" +
            "  where behavior = 'pv' " +
            "  GROUP BY window_start, window_end , itemId";

        Table t2 = tableEnvironment.sqlQuery(hopWindowSql);
        tableEnvironment.createTemporaryView("t2",t2);

        /*
            3.按照 窗口范围 分组，求每组中 所有商品的 click的排名

            排名是一个窗口函数:
                hive:   rank，dense_rank,row_number....
                flink:  row_number

                window子句不写，就是真个分组的范围。

         */
        String rankSql = " select  itemId, window_start, window_end, click ," +
            "                  row_number() over( partition by window_end order by click desc ) rn  " +
            "             from t2  ";

        Table t3 = tableEnvironment.sqlQuery(rankSql);
        tableEnvironment.createTemporaryView("t3",t3);

        //4. 按照排名 过滤出 前3
        /*
                在flink中，topN操作，只有 over()窗口，和过滤一起查询，解释器才会解释为topN操作，此时 over(  order by  任意字段)

                over()窗口，后续没有再将排名过滤，解释器不会解析为topN，此时 over(  order by  时间字段)。
         */

        /*
                建表映射，Mysql中的结果表，保存最终的结果。

                flink写入Mysql使用的是  insert xxx on duplicate update 方式
         */
        String mysqlTable = "CREATE TABLE `t4` (" +
            " `w_start` TIMESTAMP ," +
            "  `w_end` TIMESTAMP ," +
            "  `item_id` BIGINT ," +
            "  `item_count` BIGINT," +
            "  `rk` BIGINT," +
            "  PRIMARY KEY (`w_end`,`rk`) NOT ENFORCED " +
            ") WITH (" +
            "   'connector' = 'jdbc'," +
            "   'url' = 'jdbc:mysql://hadoop102:3306/221109?useSSL=false'," +
            "   'table-name' = 'hot_item' ," +
            "   'username' = 'root' , " +
            "   'password' = '000000' " +
            ")";

        tableEnvironment.executeSql(mysqlTable);
        tableEnvironment.executeSql("insert into t4 select window_start, window_end,itemId, click,rn  from t3 where rn <= 3 ");
        //tableEnvironment.sqlQuery("select * from t3  ")
                        //.execute().print();

    }
}
